package com.shujia.spark.streaming

import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.streaming.dstream.DStream

object Demo04Transform {
  def main(args: Array[String]): Unit = {
    /**
     * transform：将每个批次转换成RDD进行处理
     * 同foreachRDD类似，只不过一个有返回值，一个没有
     */
    val spark: SparkSession = SparkSession
      .builder()
      // 因为流处理需要一直接收数据，所以需要一直占用一个线程，
      // 所以至少需要2个及以上的线程才能保证实时任务正常运行
      .master("local[2]")
      .appName(this.getClass.getSimpleName.replace("$", ""))
      .getOrCreate()

    // 构建SparkStreaming环境
    val ssc: StreamingContext = new StreamingContext(spark.sparkContext, Durations.seconds(5))
    val textDS: DStream[String] = ssc.socketTextStream("master", 8888)

    // 统计单词的数量
    textDS
      .transform(rdd => {
        rdd
          .flatMap(_.split(","))
          .map(word => (word, 1))
          .reduceByKey(_ + _)
      })
      // 只需要保留java scala的结果
      .filter(kv => {
        Set("java", "scala").contains(kv._1)
      })
      .print()

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }

}
